package com.gugee.tiktok.data.consumer.blogger.core.main;

import cn.hutool.core.date.DateUtil;
import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.tiktok.common.model.dal.blogger.BloggerFansPortraitRepository;
import com.gugee.tiktok.common.model.dal.blogger.BloggerMainRepository;
import com.gugee.tiktok.common.model.dal.common.DicLanguageRepository;
import com.gugee.tiktok.common.model.dal.common.DicRegionRepository;
import com.gugee.tiktok.common.model.dal.market.BloggerMarketRepository;
import com.gugee.tiktok.common.model.es.ESBloggerFansPortrait;
import com.gugee.tiktok.common.model.es.StatisticsData;
import com.gugee.tiktok.common.model.mysql.BloggerFansPortrait;
import com.gugee.tiktok.common.model.mysql.BloggerMarket;
import com.gugee.tiktok.common.model.mysql.DicLanguage;
import com.gugee.tiktok.common.model.mysql.DicRegion;
import com.gugee.tiktok.data.common.msg.BloggerMessage;
import com.gugee.tiktok.data.consumer.blogger.config.DataConsumerConfig;
import com.xiguaji.boot.toolkit.DateTimeUtils;
import com.xiguaji.boot.toolkit.JsonUtil;
import com.xiguaji.boot.toolkit.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;


@Component
@Slf4j
public class SaveFansPortrait {

    ForkJoinPool forkJoinPool;

    @Autowired
    DataConsumerConfig dataConsumerConfig;

    private final String BLOGGER_INDEX = "blogger_main_idx";
    @Autowired
    RestHighLevelClient highLevelClient;

    /**
     * 粉丝画像 Dao
     */
    @Autowired
    BloggerFansPortraitRepository bloggerFansPortraitRepository;

    @Autowired
    BloggerMainRepository bloggerMainRepository;

    /**
     * 星图dao
     */
    @Autowired
    BloggerMarketRepository bloggerMarketRepository;
    /**
     * 语言dao
     */
    @Autowired
    DicLanguageRepository dicLanguageRepository;

    /**
     * 地区dao
     */
    @Autowired
    DicRegionRepository dicRegionRepository;

    Map<String,String>  dicLanguageMap;

    Map<String,String> dicRegionMap;

    @Autowired
    ScheduleTemplate scheduleTemplate;

    @PostConstruct
    public void construct(){
        dicLanguageMap = getDicLanguageMap();
        dicRegionMap = getDicRegionMap();
        forkJoinPool = ThreadPoolUtils.forkJoinPool(dataConsumerConfig.getFansPortraitThreadPoolSize(),"ESSaveFansPortrait");
        scheduleTemplate.attach("ESSaveFansPortrait", forkJoinPool);
    }


    public void handle(List<BloggerMessage> bloggerMessages) throws ExecutionException, InterruptedException {

        BulkRequest bulkRequest = new BulkRequest();
        //获取限制的时间范围
        int limitTime = Integer.parseInt(DateUtil.format(DateTimeUtils.localDateTimeBySecond(
                DateTimeUtils.currentTimeSecond() - dataConsumerConfig.getLatestDays() * 24 * 3600),"yyyyMMdd"));

        forkJoinPool.submit(()->
            bloggerMessages.forEach( bloggerMessage -> {
                try{
                    ESBloggerFansPortrait esBloggerFansPortrait = statisticsBloggerFansPortrait(bloggerMessage.getUid(),limitTime);

                    String jsonAweme= JsonUtil.parseJson(esBloggerFansPortrait);
                    UpdateRequest updateRequest = new UpdateRequest(BLOGGER_INDEX, String.valueOf(bloggerMessage.getId())).doc(jsonAweme,XContentType.JSON)
                            .upsert(jsonAweme,XContentType.JSON);
                    bulkRequest.add(updateRequest);
                }catch (Exception e){
                    log.info("Save Fans Portrait error",e.getMessage());
                }
            })
        ).get();

        try {
            val bulk = highLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);

        } catch (IOException e) {
            log.error("ES Save Fans Portrait error",e);
        }

    }


    /**
     * 根据视频的粉丝画像统计博主的粉丝画像
     * @param bloggerId
     * @param limitTime
     * @return
     */
    private ESBloggerFansPortrait statisticsBloggerFansPortrait(long bloggerId,int limitTime){

        List<BloggerFansPortrait> bloggerFansPortraits = bloggerFansPortraitRepository.findByUid(bloggerId,limitTime);

        ESBloggerFansPortrait bloggerFansPortraitResponse = new ESBloggerFansPortrait();
        //bloggerFansPortraitResponse.setUid(bloggerId);
        //语言
        Map<String,Integer> languageMap = new HashMap<>();
        //活跃的小时
        int[] activeHours = new int[24];
        //活跃的周内时间
        int[] activeWeekDays = new int[7];
        //地域
        Map<String,Integer> regionMap = new HashMap<>();
        //地域
        Map<String,Integer> ageMap = new HashMap<>();
        //地域
        Map<String,Integer> genderMap = new HashMap<>();
        //统计数
        int languageTotal = 0;
        int regionTotal = 0;
        int ageTotal = 0;
        int genderTotal = 0;
        List<Integer> hours;
        List<Integer> weekDays;
        Set<String> noStatParam = new HashSet<>();
        noStatParam.add("UNKNOWN");

        //统计
        for(BloggerFansPortrait bloggerFansPortrait : bloggerFansPortraits){
            //语言分布
            languageTotal += statisticsParam(languageMap,bloggerFansPortrait.getLanguageDistribution(),null);
            //地区分布
            regionTotal += statisticsParam(regionMap,bloggerFansPortrait.getRegionDistribution(),null);
            //活跃小时
            hours = JsonUtil.jsonParse(List.class,bloggerFansPortrait.getActiveDayDistribution());
            if(hours != null && hours.size() > 0){
                for(int i = 0; i < activeHours.length && i < hours.size(); ++i){
                    activeHours[i] += hours.get(i);
                }
            }
            //活跃日期
            weekDays = JsonUtil.jsonParse(List.class,bloggerFansPortrait.getActiveWeekDistribution());
            if(weekDays != null && weekDays.size() > 0){
                for(int i = 0; i < activeWeekDays.length && i < weekDays.size(); ++i){
                    activeWeekDays[i] += weekDays.get(i);
                }
            }
            //性别
            genderTotal += statisticsParam(genderMap,bloggerFansPortrait.getGenderDistribution(),noStatParam);
            //年龄
            ageTotal += statisticsParam(ageMap,bloggerFansPortrait.getAgeDistribution(),noStatParam);
            //总数
            // totalComment += bloggerFansPortrait.getCommentsSampleCount();
        }
        //汇总实体类
        bloggerFansPortraitResponse.setActiveDayDistribution(getPercentageByStat(activeHours));
        bloggerFansPortraitResponse.setActiveWeekDistribution(getPercentageByStat(activeWeekDays));
        bloggerFansPortraitResponse.setLanguageDistribution(transMap2StatisticsData(languageMap,
                dicLanguageMap,languageTotal,dataConsumerConfig.getFansPortraitStatisticsLanguageTop(),"语言"));
        bloggerFansPortraitResponse.setRegionDistribution(transMap2StatisticsData(regionMap,
                dicRegionMap,regionTotal,dataConsumerConfig.getFansPortraitStatisticsRegionTop(),"国家"));
        if(!genderMap.isEmpty()){
            bloggerFansPortraitResponse.setGenderDistribution(getAgeAndGenderPercentageByStat(genderMap,genderTotal));
            bloggerFansPortraitResponse.setAgeDistribution(getAgeAndGenderPercentageByStat(ageMap,ageTotal));
        }else{
            //查询星图
            val bloggerMarket = bloggerMarketRepository.findByCoreUserId(bloggerId);
            if(bloggerMarket.isPresent()){
                BloggerMarket bm = bloggerMarket.get();
                // bloggerFansPortraitResponse.setActive(bm.getActive());
                try{
                    List<StatisticsData<Double>> gender = transJsonString2StatisticsData(bm.getGenderDistribution());
                    bloggerFansPortraitResponse.setGenderDistribution(gender);
                }catch (Exception e){
                    log.error("粉丝画像-星图性别分布json转换失败 uid: {} ,genderString: {}",bloggerId,bm.getGenderDistribution());
                }
                try{
                    List<StatisticsData<Double>> age = transJsonString2StatisticsData(bm.getAgeDistribution());
                    bloggerFansPortraitResponse.setAgeDistribution(age);
                }catch (Exception e){
                    log.error("粉丝画像-星图年龄分布json转换失败 uid: {} ,genderString: {}",bloggerId,bm.getGenderDistribution());
                }
            }
        }

        // bloggerFansPortraitResponse.setCommentsSampleCount(totalComment);
        //更新粉丝画像统计时间
        bloggerMainRepository.updateBloggerFansPortraitUpdate(new Date().getTime()/1000,bloggerId);
        bloggerFansPortraitResponse.setFansPortraitUpdate(new Date().getTime()/1000);

        // 性别占比最大的
        bloggerFansPortraitResponse.setGenderDistributionMax(
                bloggerFansPortraitResponse.getGenderDistribution().stream().max((a,b)->a.getValue() > b.getValue()? 1:-1).get().getProperty());
        // 年龄占比最大的
        bloggerFansPortraitResponse.setAgeDistributionMax(
                bloggerFansPortraitResponse.getAgeDistribution().stream().max((a,b)->a.getValue() > b.getValue()? 1:-1).get().getProperty());
        // 语言占比最大的
        bloggerFansPortraitResponse.setLanguageDistributionMax(
                bloggerFansPortraitResponse.getLanguageDistribution().stream().max((a,b)->a.getValue() > b.getValue()? 1:-1).get().getProperty());
        // 国家占比最大的
        bloggerFansPortraitResponse.setRegionDistributionMax(
                bloggerFansPortraitResponse.getRegionDistribution().stream().max((a,b)->a.getValue() > b.getValue()? 1:-1).get().getProperty());
        return bloggerFansPortraitResponse;
    }

    /**
     * 将星图json字符串转换成统计数据
     * @param jsonString 星图json字符串
     * @return
     */
    private List<StatisticsData<Double>> transJsonString2StatisticsData (String jsonString){
        List<Map<String,Object>> list = JsonUtil.jsonParse(List.class,jsonString);
        List<StatisticsData<Double>> result;
        //除去最后一个值的百分比
        long withoutLastPercentageVale = 0;
        long value = 0;
        if(list != null && list.size() > 0){
            result = new ArrayList<>(list.size());
            for(Map<String,Object> temp : list){
                if(temp != null && temp.get("property") != null){
                    StatisticsData<Double> stat = new StatisticsData<>();
                    stat.setProperty((String)temp.get("property"));
                    value = Math.round((double)temp.get("value") * 10000);
                    withoutLastPercentageVale += value;
                    stat.setValue((double)value / 100);
                    result.add(stat);
                }
            }
            withoutLastPercentageVale -= value;
            if(result.size() > 1){
                result.get(result.size() - 1).setValue((double)(10000 - Math.round(withoutLastPercentageVale)) / 100);
            }
        }else{
            result = new ArrayList<>();
        }
        return result;
    }

    /**
     * 统计某项参数
     * @param total 某项参数的合计
     * @param paramJson 当个视频中的某项参数
     * @param noStatParam 不用统计的参数名
     */
    private int statisticsParam(Map<String,Integer> total,String paramJson,Set<String> noStatParam){
        int count = 0;
        if(!StringUtils.isEmpty(paramJson)){
            Map<String,Integer> temp = JsonUtil.jsonParse(Map.class,paramJson);
            if(temp != null && !temp.isEmpty()){
                Iterator<Map.Entry<String,Integer>> iterator = temp.entrySet().iterator();
                Map.Entry<String,Integer> entry;
                while(iterator.hasNext()){
                    entry = iterator.next();
                    if(noStatParam == null || noStatParam.isEmpty() || !noStatParam.contains(entry.getKey())){
                        count += entry.getValue();
                        total.put(entry.getKey(),total.getOrDefault(entry.getKey(),0) + entry.getValue());
                    }
                }
            }
        }
        return count;
    }

    /**
     * 将活跃天数的数据转换成百分比
     * @param dateTimeStat 统计的日期数据
     * @return
     */
    private double[] getPercentageByStat(int[] dateTimeStat){
        double[] result = new double[dateTimeStat.length];
        int sum = 0;
        for(int stat : dateTimeStat){
            sum += stat;
        }
        if(sum > 0){
            for(int i = 0; i < dateTimeStat.length; ++i){
                result[i] = (double)((long)dateTimeStat[i] * 10000 / sum) / 100;
            }
        }
        return result;
    }

    /**
     * 统计数据转换成百分比排序集
     * @param paramMap 原统计数据
     * @param dicMap 语言字典
     * @param total 评论样本数
     * @param topNum 前几个
     * @param dicName 字典类型
     * @return
     */
    private List<StatisticsData<Double>> transMap2StatisticsData(Map<String,Integer> paramMap,
                                                                 Map<String,String> dicMap,int total,int topNum,
                                                                 String dicName){
        List<StatisticsData<Double>> result = new ArrayList<>(topNum);
        if(!paramMap.isEmpty()){
            //优先队列排序
            PriorityQueue<StatisticsData<Double>> priorityQueue = new PriorityQueue<>(paramMap.size(),(o1,o2) -> -(o1.getValue().compareTo(o2.getValue())));

            //map 转 list
            Iterator<Map.Entry<String,Integer>> iterator = paramMap.entrySet().iterator();

            Map.Entry<String,Integer> entry;
            while(iterator.hasNext()){
                entry = iterator.next();
                StatisticsData<Double> stat = new StatisticsData<>();
                stat.setProperty(entry.getKey());
                stat.setValue((double)(entry.getValue().longValue() * 10000 / total) / 100);
                priorityQueue.add(stat);
            }
            StatisticsData<Double> stat;
            String dicValue;
            while(!priorityQueue.isEmpty() && topNum > 0){
                --topNum;
                stat = priorityQueue.poll();
                stat.getProperty();
                dicValue = dicMap.get(stat.getProperty());
                if(StringUtils.isEmpty(dicValue)){
                    log.error("粉丝画像统计--未找到 {} 字典值 {}",dicName,stat.getProperty());
                }else{
                    stat.setProperty(dicValue);
                }
                result.add(stat);
            }
        }
        return result;
    }

    /**
     * 将年龄段、性别数据的数据转换成百分比
     * @param totalMap
     * @param total
     * @return
     */
    private List<StatisticsData<Double>> getAgeAndGenderPercentageByStat (Map<String,Integer> totalMap,int total){
        List<StatisticsData<Double>> result = new ArrayList<>(totalMap.size());
        //除去最后一个值的百分比
        if(!totalMap.isEmpty() && total > 0){
            long withoutLastPercentageVale = 0;
            long value = 0;
            Iterator<Map.Entry<String,Integer>> iterator = totalMap.entrySet().iterator();
            Map.Entry<String,Integer> entry;
            //遍历分析
            while(iterator.hasNext()){
                entry = iterator.next();
                StatisticsData<Double> stat = new StatisticsData<>();
                stat.setProperty(entry.getKey());
                value = entry.getValue().longValue() * 10000 / total;
                withoutLastPercentageVale += value;
                stat.setValue((double)value / 100);
                result.add(stat);
            }
            withoutLastPercentageVale -= value;
            if(result.size() > 1){
                result.get(result.size() - 1).setValue((double)(10000 - Math.round(withoutLastPercentageVale)) / 100);
            }
        }
        return result;
    }

    /**
     * 获取语言字典
     * @return
     */
    private Map<String,String> getDicLanguageMap(){
        Map<String,String> dicMap;
        val dicLanguages = dicLanguageRepository.findAll();
        Iterator<DicLanguage> dicLanguageList = dicLanguages.iterator();
        dicMap = new HashMap<>();
        DicLanguage dicLanguage;
        while (dicLanguageList.hasNext()){
            dicLanguage = dicLanguageList.next();
            if(dicLanguage != null && org.apache.commons.lang3.StringUtils.isNotEmpty(dicLanguage.getCode())){
                dicMap.put(dicLanguage.getCode().toLowerCase(),dicLanguage.getName());
            }
        }
        return dicMap;
    }

    /**
     * 获取地区字典
     * @return
     */
    private Map<String,String> getDicRegionMap(){
        Map<String,String> dicMap;
        val dicLanguages = dicRegionRepository.findAll();
        Iterator<DicRegion> dicLanguageList = dicLanguages.iterator();
        dicMap = new HashMap<>();
        DicRegion region;
        while (dicLanguageList.hasNext()){
            region = dicLanguageList.next();
            if(region != null && org.apache.commons.lang3.StringUtils.isNotEmpty(region.getCode())){
                dicMap.put(region.getCode().toLowerCase(),region.getName());
            }
        }
        return dicMap;
    }
}
